package cn.tedu.charging.device.canal;

import cn.tedu.charging.device.pojo.canal.po.StationCanalPO;
import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import jakarta.annotation.PostConstruct;
import jakarta.annotation.PreDestroy;
import java.math.BigDecimal;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * 场站数据 Canal 监听服务
 * 使用阿里巴巴官方 canal.client 实现数据同步
 */
@Slf4j
@Service
public class StationCanalService {

    @Autowired
    private CanalConnector canalConnector;

    private ExecutorService executorService;

    private volatile boolean running = false;

    @PostConstruct
    public void start() {
        executorService = Executors.newSingleThreadExecutor();
        running = true;
        executorService.submit(new CanalWorker());
        log.info("Canal监听服务已启动");
    }

    @PreDestroy
    public void stop() {
        running = false;
        if (executorService != null) {
            executorService.shutdown();
        }
        log.info("Canal监听服务已停止");
    }

    private class CanalWorker implements Runnable {
        @Override
        public void run() {
            while (running) {
                try {
                    // 连接 Canal Server
                    canalConnector.connect();
                    // 订阅 charging_station 表
                    canalConnector.subscribe(".*\\.charging_station");
                    // 回滚到未 ack 的位置
                    canalConnector.rollback();

                    while (running) {
                        // 获取消息，每次最多获取1000条记录
                        Message message = canalConnector.getWithoutAck(1000);
                        long batchId = message.getId();
                        int size = message.getEntries().size();

                        if (batchId == -1 || size == 0) {
                            // 没有数据时休眠2秒
                            Thread.sleep(2000);
                        } else {
                            // 处理数据
                            processEntries(message.getEntries());
                            // 提交确认
                            canalConnector.ack(batchId);
                        }
                    }
                } catch (Exception e) {
                    log.error("Canal数据处理异常", e);
                    try {
                        Thread.sleep(5000); // 发生异常时休眠5秒再重试
                    } catch (InterruptedException ie) {
                        Thread.currentThread().interrupt();
                    }
                } finally {
                    canalConnector.disconnect();
                }
            }
        }
    }

    /**
     * 处理 Canal 条目数据
     * @param entries Canal 条目列表
     */
    private void processEntries(List<CanalEntry.Entry> entries) {
        for (CanalEntry.Entry entry : entries) {
            if (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONBEGIN 
                || entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONEND) {
                continue;
            }

            try {
                CanalEntry.RowChange rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
                CanalEntry.EventType eventType = rowChange.getEventType();

                for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) {
                    switch (eventType) {
                        case INSERT:
                            handleInsert(rowData);
                            break;
                        case UPDATE:
                            handleUpdate(rowData);
                            break;
                        case DELETE:
                            handleDelete(rowData);
                            break;
                        default:
                            break;
                    }
                }
            } catch (Exception e) {
                log.error("解析Canal数据异常", e);
            }
        }
    }

    /**
     * 处理插入操作
     * @param rowData 行数据
     */
    private void handleInsert(CanalEntry.RowData rowData) {
        StationCanalPO station = convertToStation(rowData.getAfterColumnsList());
        log.debug("监听到新增数据:{}", station);
        log.debug("新增数据开始同步到Redis/ES/MQ");
        // TODO: 通过RedisTemplate 写入数据
        log.debug("数据库新增的数据Redis同步新增成功");
    }

    /**
     * 处理更新操作
     * @param rowData 行数据
     */
    private void handleUpdate(CanalEntry.RowData rowData) {
        StationCanalPO before = convertToStation(rowData.getBeforeColumnsList());
        StationCanalPO after = convertToStation(rowData.getAfterColumnsList());
        log.debug("监听到更新数据:before:{},after:{}", before, after);
        log.debug("更新数据开始同步到Redis");
        // TODO: 通过RedisTemplate 更新数据
        log.debug("数据库更新的数据Redis同步更新成功");
    }

    /**
     * 处理删除操作
     * @param rowData 行数据
     */
    private void handleDelete(CanalEntry.RowData rowData) {
        StationCanalPO station = convertToStation(rowData.getBeforeColumnsList());
        log.debug("监听到删除数据:{}", station);
        log.debug("删除数据开始同步到Redis");
        // TODO: 通过RedisTemplate 删除数据
        log.debug("数据库删除的数据Redis同步删除成功");
    }

    /**
     * 将 Canal 列数据转换为 StationCanalPO 对象
     * @param columns 列数据列表
     * @return StationCanalPO 对象
     */
    private StationCanalPO convertToStation(List<CanalEntry.Column> columns) {
        StationCanalPO station = new StationCanalPO();
        for (CanalEntry.Column column : columns) {
            switch (column.getName()) {
                case "id":
                    if (column.getValue() != null && !column.getValue().isEmpty()) {
                        station.setId(Integer.valueOf(column.getValue()));
                    }
                    break;
                case "station_name":
                    station.setStationName(column.getValue());
                    break;
                case "station_lng":
                    if (column.getValue() != null && !column.getValue().isEmpty()) {
                        station.setStationLng(new BigDecimal(column.getValue()));
                    }
                    break;
                case "station_lat":
                    if (column.getValue() != null && !column.getValue().isEmpty()) {
                        station.setStationLat(new BigDecimal(column.getValue()));
                    }
                    break;
                default:
                    break;
            }
        }
        return station;
    }
}